package cn.doitedu.demo.connectors;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Jdbc连接器_读 {
    public static void main(String[] args) {
        // 构造stream api 编程的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        // 构造 sql 编程环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 创建目标数据源映射表，映射mysql中的表
        tenv.executeSql("CREATE TABLE t2                 ( \n" +
                "  id int,                                     \n" +
                "  name  string,                               \n" +
                "  gender  string,                              \n" +
                "  score  double,                              \n" +
                "  primary key(id) not enforced               \n" +
                ") WITH (                                    \n" +
                "   'connector' = 'jdbc',                    \n" +
                "   'url' = 'jdbc:mysql://doitedu:3306/flinktest',\n" +
                "   'table-name' = 'flink_score',                   \n" +
                "   'username' = 'root',                      \n" +
                "   'password' = 'root'                       \n" +
                ");");

        tenv.executeSql("select * from t2").print();









    }
}
